package com.jiangwei.flv.factories;

import com.alibaba.fastjson.util.IOUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.catalina.connector.ClientAbortException;
import org.apache.coyote.CloseNowException;
import org.bytedeco.ffmpeg.avcodec.AVPacket;
import org.bytedeco.ffmpeg.global.avcodec;
import org.bytedeco.javacv.FFmpegFrameGrabber;
import org.bytedeco.javacv.FFmpegFrameRecorder;

import javax.servlet.AsyncContext;
import javax.servlet.ServletOutputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static org.bytedeco.ffmpeg.global.avutil.AV_NOPTS_VALUE;

/**
 * javacv转包装<br/>
 * 无须转码，更低的资源消耗，更低的延迟<br/>
 * 确保流来源视频H264格式,音频AAC格式
 *
 * @author gc.x
 */
@Slf4j
public class ConverterFactories implements Converter, Runnable {
    public volatile boolean running = true;
    /**
     * 读流器
     */
    private FFmpegFrameGrabber grabber;
    /**
     * 转码器
     */
    private FFmpegFrameRecorder recorder;
    /**
     * 转FLV格式的头信息<br/>
     * 如果有第二个客户端播放首先要返回头信息
     */
    private byte[] headers;
    /**
     * 保存转换好的流
     */
    private ByteArrayOutputStream stream;
    /**
     * 流地址，h264,aac
     */
    private final String url;
    /**
     * 流输出
     */
    private final List<AsyncContext> outEntities;

    /**
     * key用于表示这个转换器
     */
    private final String key;

    /**
     * 转换队列
     */
    private final Map<String, Converter> factories;

    public ConverterFactories(String url, String key, Map<String, Converter> factories, List<AsyncContext> outEntities, FFmpegFrameGrabber grabber) {
        this.url = url;
        this.key = key;
        this.factories = factories;
        this.outEntities = outEntities;
        this.grabber = grabber;
    }

    @Override
    public void run() {
        try {
            log.info("this url:{} converterFactories start", url);
            // 来源视频H264格式,音频AAC格式
            // 无须转码，更低的资源消耗，更低的延迟
            stream = new ByteArrayOutputStream();
            recorder = new FFmpegFrameRecorder(stream, grabber.getImageWidth(), grabber.getImageHeight(),
                    grabber.getAudioChannels());
            recorder.setInterleaved(true);
            recorder.setVideoOption("preset", "ultrafast");
            recorder.setVideoOption("tune", "zerolatency");
            recorder.setVideoOption("crf", "25");
            recorder.setFrameRate(grabber.getFrameRate());
            recorder.setSampleRate(grabber.getSampleRate());
            if (grabber.getAudioChannels() > 0) {
                recorder.setAudioChannels(grabber.getAudioChannels());
                recorder.setAudioBitrate(grabber.getAudioBitrate());
                recorder.setAudioCodec(grabber.getAudioCodec());
            }
            recorder.setFormat("flv");
            recorder.setVideoBitrate(grabber.getVideoBitrate());
            recorder.setVideoCodec(grabber.getVideoCodec());
            recorder.start(grabber.getFormatContext());
            if (headers == null) {
                headers = stream.toByteArray();
                stream.reset();
                writeResponse(headers);
            }
            long startTime = System.currentTimeMillis();
            long videoTS;
            long lastDTS = 0;
            while (running) {
                AVPacket packet = grabber.grabPacket();
                if (packet != null) {
                    if (packet.pts() == AV_NOPTS_VALUE) {
                        if (packet.dts() != AV_NOPTS_VALUE) {
                            packet.pts(packet.dts());
                            lastDTS = packet.dts();
                        } else {
                            packet.pts(lastDTS + 1);
                            packet.dts(packet.pts());
                            lastDTS = packet.pts();
                        }
                    } else {
                        if (packet.dts() != AV_NOPTS_VALUE) {
                            if (packet.dts() < lastDTS) {
                                packet.dts(lastDTS + 1);
                            }
                        } else {
                            packet.dts(lastDTS + 1);
                        }
                        lastDTS = packet.dts();
                    }
                    if (packet.pts() < packet.dts()) {
                        packet.pts(packet.dts());
                    }
                    videoTS = 1000 * (System.currentTimeMillis() - startTime);
                    if (videoTS < 0 || packet.dts() < 0 || packet.pts() < 0) {
                        continue;
                    }
                    if (videoTS > recorder.getTimestamp()) {
                        recorder.setTimestamp(videoTS);
                    }
                    recorder.recordPacket(packet);
                    if (stream.size() > 0) {
                        byte[] b = stream.toByteArray();
                        stream.reset();
                        writeResponse(b);
                        if (outEntities.isEmpty()) {
                            log.info("没有输出退出");
                            break;
                        }
                    }
                    avcodec.av_packet_unref(packet);
                } else {
                    grabber = new FFmpegFrameGrabber(url);
                    grabber.setOption("rtsp_transport", "tcp");
                    grabber.setOption("stimeout", "5000000");
                    grabber.start();
                }
                TimeUnit.MILLISECONDS.sleep(5);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            closeConverter();
            completeResponse();
            log.info("this url:{} converterTranFactories exit", url);
            factories.remove(this.key);
        }
    }

    /**
     * 输出FLV视频流
     */
    public void writeResponse(byte[] b) {
        Iterator<AsyncContext> it = outEntities.iterator();
        while (it.hasNext()) {
            AsyncContext o = it.next();
            try {
                ServletOutputStream outputStream = o.getResponse().getOutputStream();
                outputStream.write(b);
            } catch (ClientAbortException | CloseNowException e) {
                // log.info("移除一个输出");
                it.remove();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 退出转换
     */
    public void closeConverter() {
        IOUtils.close(grabber);
        IOUtils.close(recorder);
        IOUtils.close(stream);
    }

    /**
     * 关闭异步响应
     */
    public void completeResponse() {
        for (AsyncContext o : outEntities) {
            o.complete();
        }
    }

    @Override
    public String getKey() {
        return this.key;
    }

    @Override
    public String getUrl() {
        return this.url;
    }

    @Override
    public void addOutputStreamEntity(String key, AsyncContext entity) throws IOException {
        if (headers == null) {
            outEntities.add(entity);
        } else {
            entity.getResponse().getOutputStream().write(headers);
            entity.getResponse().getOutputStream().flush();
            outEntities.add(entity);
        }
    }

    @Override
    public void exit() {
        this.running = false;
//		try {
//			this.join();
//		} catch (Exception e) {
//			log.error(e.getMessage(), e);
//		}
    }

    @Override
    public void setFlag() {

    }

}
